package gspushmq.v1.consumer.core;

import static gspushmq.v1.consumer.core.Validate.validate;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

import gspushmq.model.*;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.WebSocket;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import ggframework.mongodb.BasicDBObject;
import ggframework.mongodb.DBCollection;
import ggframework.mongodb.DBCursor;
import ggframework.mongodb.DBObject;
import gspushmq.util.MongoUtil;
import gspushmq.v1.save.SaveSendRecord;
import gspushmq.v1.subscribe.GGSubscribe;
import gspushmq.v1.thread.GGThreadPool;

/**
 * @Description PC推送wss工具类
 * @Author 成尚佳
 * @Date 2019/1/9 10:44
 * @Version 1.0
 * @company 上海朝阳永续信息技术有限公司
 * @copyright (c) 2015 SunTime Co'Ltd Inc. All rights reserved.
 **/
public class PushCore {


    /**
     * @param user
     * @param ws
     * @param mqSubTypes
     * @description 推送离线消息给登录用户
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 16:14
     */

    public static void sendLoginUser(User user, WebSocket ws, List<String> mqSubTypes) {
        if (validate(user)) {
            Integer productLine = Integer.valueOf(user.getProductLine());
            Long accounId = Long.valueOf(user.getAccountId());
            String accountName = user.getAccountName();
            Integer pro = user.getProType();
            Set<Data> noExpredMessage = listNoExpiredMessages(accounId, accountName, productLine, pro, mqSubTypes);
            Set<Data> noECMessage = listNoConsumeredAndNoExpiredMessages(noExpredMessage, productLine, accounId);
            if (noECMessage != null && noECMessage.size() > 0) {
                if (ws.isOpen()) {
                    for (Data data : noECMessage) {
                        consumer(data, user, ws);
                    }
                }
            }
        }
    }

    /**
     * @param data
     * @param user
     * @param ws
     * @description 支持的消息类型
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/22 10:32
     */
    private static void consumer(Data data, User user, WebSocket ws) {
        //todo 下版本去掉特殊结构，采用统一格式的返回数据

        if (TempRequestIdEnum.TAGS.getMqSubType().equals(data.getMqSubType())) {
            //标签异动
            sendWS(data, ws);
            SaveSendRecord.save(data, user);
        } else {

            String reqId = TempRequestIdEnum.getRequestId(data.getMqSubType());
            if (StringUtils.isNotBlank(reqId)) {
                sendWS(reqId, data, ws);
                SaveSendRecord.save(data, user);
            }
        }
    }

    /**
     * @param data
     * @description 推送消息给在线用户
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 14:30
     */
    public static void sendOnlineUsers(Data data) {
        if (data == null) {
            return;
        }
        for (Map.Entry<WebSocket, User> webSocketUserEntry : GGSubscribe.wstUser.entrySet()) {
            WebSocket ws = webSocketUserEntry.getKey();
            User user = webSocketUserEntry.getValue();
            if (matchSingleUser(data, user)) {
                if (ws.isOpen()) {
                    consumer(data, user, ws);
                }

            }
        }
    }


    /////////////////////////////////////////////////////////////////////////////////////////

    /**
     * @param messageSet
     * @return set<String>
     * @description 获取uuid集合
     * @version 1.0
     * @author chengsja
     * @date 2018/12/14 15:00
     */
    private static Set<String> getUUIDs(Set<Data> messageSet) {
        if (messageSet == null || messageSet.size() == 0) {
            return null;
        }
        Set<String> set = Sets.newHashSet();
        for (Data dataMessage : messageSet) {
            set.add(dataMessage.getUuid());
        }
        return set;
    }


    /**
     * @param pdm
     * @param productLine 产品线
     * @param accountId   账户id(消费表中核心字段)
     * @description 获取产品线下未过期未消费的消息集合
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 15:52
     */
    private static Set<Data> listNoConsumeredAndNoExpiredMessages(Set<Data> pdm, Integer productLine, Long accountId) {
        Set<String> puuids = getUUIDs(pdm);
        if (puuids == null || puuids.size() == 0 || productLine == null || accountId == null) {
            return null;
        }
        if (puuids.size() > 0) {
            DBCollection record = MongoUtil.getFtAccountBehaviorCollection("cabs_msg_send_record");
            DBCursor cursor = record.find(new BasicDBObject("uuid", new BasicDBObject("$in", puuids))
                    .append("product_line", productLine)
                    .append("account_id", accountId), new BasicDBObject("uuid", 1));
            Set<Data> delDm = Sets.newHashSet();
            while (cursor.hasNext()) {
                DBObject o = cursor.next();
                for (Data dm : pdm) {
                    if (o.getString("uuid").equals(dm.getUuid())) {
                        delDm.add(dm);
                    }
                }
            }
            cursor.close();
            //过滤已推送的
            pdm.removeAll(delDm);
        }
        return pdm;
    }

    /**
     * @param accountId   账号id，匹配alias类型
     * @param accountName 账号名，匹配account类型
     * @param productLine 产品线，匹配产品线
     * @param pro         权限，匹配tag类型
     * @param mqSubTypes  消息详细类型，匹配消息子类型
     * @description 获取未过期的消息集合
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 15:47
     */
    private static Set<Data> listNoExpiredMessages(Long accountId, String accountName, Integer productLine, Integer pro, List<String> mqSubTypes) {

        //查生产库
        if (accountId == null || productLine == null || pro == null
                || accountName == null || mqSubTypes == null || mqSubTypes.size() == 0) {
            return null;
        }
        List<DBObject> orList = new ArrayList<>();
        //1.获取用户终端类型，匹配推送数据为TAG推送
        Integer tag = VersionEnum.getNumber(productLine, Integer.valueOf(pro));
        if (tag != null) {
            DBObject or1 = new BasicDBObject();
            or1.put("crowdType", "TAG");
            or1.put("crowdValue", new BasicDBObject("$regex", String.valueOf(tag)));
            orList.add(or1);
        }
        //2.获取用户名，匹配推送数据为ACCOUNT推送
        if (StringUtils.isNotBlank(accountName)) {
            DBObject or2 = new BasicDBObject();
            or2.put("crowdType", "ACCOUNT");
            or2.put("crowdValue", new BasicDBObject("$regex", accountName));
            orList.add(or2);
        }
        //3.获取账号，匹配推送数据为ALAS推送
        if (accountId != null) {
            DBObject or3 = new BasicDBObject();
            or3.put("crowdType", "ALIAS");
            or3.put("crowdValue", new BasicDBObject("$regex", String.valueOf(accountId)));
            orList.add(or3);
        }
        //4.所有用户消息
        DBObject or4 = new BasicDBObject();
        or4.put("crowdType", "ALL");
        orList.add(or4);

        //获取用户注册之后的系统消息
        DBCollection temCollection = MongoUtil.getFtAccountCollection("tau_account_info");
        DBObject userInfo = temCollection.findOne(new BasicDBObject("status",1).append("account_id",accountId),new BasicDBObject("insert_time",1));
        Date userRegistDate;
        if (userInfo == null || (userRegistDate =userInfo.getDate("insert_time")) == null){
            //获取失败
            return null;
        }

        DBCollection messRecord = MongoUtil.getFtAccountBehaviorCollection("cbs_msg_storage_record");
        DBObject query = new BasicDBObject();
        query.put("insert_time",new BasicDBObject("$gte",userRegistDate));
        query.put("mqSubType", new BasicDBObject("$in", mqSubTypes));
        query.put("expireDate", new BasicDBObject("$gte", new Date()));

        //产品线（后台生产消息可带产品线，也可不带产品线，不带产品线标识推送所有产品线）
        query.put("productLine", new BasicDBObject("$in", Lists.newArrayList(null, productLine)));

        query.put("$or", orList);

        DBCursor cursor = messRecord.find(query);
        Set<Data> pdm = Sets.newHashSet();
        while (cursor.hasNext()) {
            DBObject o = cursor.next();
            Data dm = new DataMessage();
            dm.setUuid(o.getString("uuid"));
            dm.setMqType(o.getString("mqType"));
            dm.setMqSubType(o.getString("mqSubType"));
            dm.setTitle(o.getString("title"));
            dm.setContent(o.get("content"));
            dm.setProductLine(o.getInteger("productLine"));
            dm.setFileType(o.getString("fileType"));
            dm.setFileAddress(o.getString("fileAddress"));
            dm.setLinkType(o.getString("linkType"));
            dm.setLinkUrl(o.getString("linkUrl"));
            dm.setModule(o.getString("module"));
            dm.setTarget(o.getString("target"));
            dm.setCrowdType(o.getString("crowdType"));
            dm.setCrowdValue(o.getString("crowdValue"));
            dm.setStartDate(o.getDate("startDate"));
            dm.setExpireDate(o.getDate("expireDate"));
            pdm.add(dm);
        }
        cursor.close();
        return pdm;
    }


    /////////////////////////////////////////////////////////////////////////////////////////

    /**
     * @param data
     * @param user
     * @description 根据消息匹配是否符合当前User对象
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 14:06
     */
    private static boolean matchSingleUser(Data data, User user) {
        if (validate(data) && validate(user)) {
            return matchProduceAndConsumer(data.getCrowdType()
                    , data.getCrowdValue()
                    , data.getProductLine()
                    , user.getProType()
                    , user.getAccountId()
                    , user.getAccountName()
                    , Integer.valueOf(user.getProductLine())
            );
        }
        return false;
    }

    /**
     * @param data
     * @description 根据消息匹配是否符合当前message对象
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 10:47
     */
    private static boolean matchSingleMessage(Data data, Message message) {
        if (validate(data) && validate(message)) {
            //后端产品线
            Integer dproductLine = data.getProductLine();
            //前端产品线
            Integer mproductLine = Integer.valueOf(message.getProduct_line());
            //前度权限
            Integer mpro = Integer.valueOf(message.getPro_type());
            //后端推送方式
            String target = data.getCrowdType();
            //后端推送值
            String targetValue = data.getCrowdValue();
            //前端账号
            String faccountId = message.getAccount_id();
            //前端账号名
            String faccountName = message.getAccount_name();
            //匹配生产消费
            return matchProduceAndConsumer(target, targetValue, dproductLine, mpro, faccountId, faccountName, mproductLine);
        }
        return false;
    }

    /**
     * @param target       生产消息-推送方式字符串
     * @param targetValue  生产消息-推送对象字符串使用；分号链接
     * @param productLine  生产消息-产品线
     * @param fpro         消费消息-前端用户权限
     * @param faccountId   消费消息-前端用户账号
     * @param faccountName 消费消息-前端用户名称
     * @param fproductLine 消费消息-前端用户产品线
     * @description 根据生产消息参数匹配消费参数
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 13:58
     */
    private static boolean matchProduceAndConsumer(String target, String targetValue, Integer productLine,
                                                   Integer fpro, String faccountId,
                                                   String faccountName, Integer fproductLine) {
        // 初始化flag
        Boolean flag = false;
        //前端消费凭证字符串
        String ftargetValue = null;


        //产品线验证（若后台消息带有产品线，则前端必须与之相匹配才行）
        if (productLine != null && !productLine.equals(fproductLine)) {
            return flag;
        }

        //推送对象解析（后端与前端的一套约定逻辑）
        if (ConsumerWays.TAG.getWay().equals(target)) {
            ftargetValue = String.valueOf(VersionEnum.getNumber(fproductLine, fpro));
        } else if (ConsumerWays.ACCOUNT.getWay().equals(target)) {
            //account推送
            //用户名
            ftargetValue = faccountName;
        } else if (ConsumerWays.ALIAS.getWay().equals(target)) {
            //id推送
            ftargetValue = faccountId;
        } else if (ConsumerWays.ALL.getWay().equals(target)) {
            //所有在线ws都推送
            ftargetValue = ConsumerWays.ALL.getWay();
        }
        // 忽略账户大小写,验证是否符合后台设置的推送对象
        if (StringUtils.isNotBlank(ftargetValue)) {
            for (String btv : targetValue.split(";")) {
                if (ftargetValue.equalsIgnoreCase(btv)) {
                    flag = true;
                    break;
                }
            }
        }
        return flag;
    }

    /**
     * @param requestId 客户端客户端请求头
     * @param data
     * @param ws
     * @description 推送消息给客户端
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/9 14:25
     */
    public static void sendWS(String requestId, Data data, WebSocket ws) {
        Map<String, Object> sMap = Maps.newHashMap();
        sMap.put("code", 0);
        sMap.put("request_id", requestId);
        sMap.put("uuid", data.getUuid());
        sMap.put("data", new Data(data));
        ws.send(JSON.toJSONString(sMap));
    }


    /**
     * @param data
     * @param ws
     * @description todo 特殊推送（标签异动）
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/15 16:18
     */
    public static void sendWS(Data data, WebSocket ws) {
        Map<String, Object> sMap = Maps.newHashMap();
        sMap.put("code", 0);
        sMap.put("request_id", TempRequestIdEnum.TAGS.getRequestId());
        sMap.put("uuid", data.getUuid());
        Object content = data.getContent();
        List<Map> maps = JSON.parseArray(content.toString(), Map.class);
        System.out.println(maps);
        sMap.put("data", maps);
        ws.send(JSON.toJSONString(sMap));
    }

}
